package com.ymm.spark.config;

import com.ymm.spark.constant.Constants;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * @author zhangj
 * @date 2019/10/26
 * JDBC辅助组件
 */
@Slf4j
public class JdbcHelper {

	static {
		String driver = ConfigurationManager.getProperty(Constants.JDBC_DRIVER);
		try {
			Class.forName(driver);
		} catch (ClassNotFoundException e) {
			log.error("JDBC加载数据库驱动异常:{}", e.getMessage(), e);
		}
	}

	/**
	 * 实例单例化
	 */
	private static JdbcHelper instance = null;

	/**
	 * 获取单例实例
	 *
	 * @return
	 */
	public static JdbcHelper getInstance() {
		if (instance == null) {
			synchronized (JdbcHelper.class) {
				if (instance == null) {
					instance = new JdbcHelper();
				}
			}
		}
		return instance;
	}

	/**
	 * 数据库连接池
	 */
	private LinkedList<Connection> datasource = new LinkedList<Connection>();

	private JdbcHelper() {
		//数据库连接池大小
		Integer datasourceSize = Integer.valueOf(ConfigurationManager.getProperty(Constants.JDBC_DATASOURCE_SIZE));
		final Boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
		String url = null;
		String username = null;
		String password = null;
		if (local) {
			url = ConfigurationManager.getProperty(Constants.JDBC_URL);
			username = ConfigurationManager.getProperty(Constants.JDBC_USER);
			password = ConfigurationManager.getProperty(Constants.JDBC_PASSWORD);
		} else {
			url = ConfigurationManager.getProperty(Constants.JDBC_URL_PROD);
			username = ConfigurationManager.getProperty(Constants.JDBC_USER_PROD);
			password = ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD);
		}
		for (int i = 0; i < datasourceSize; i++) {

			try {
				Connection conn = DriverManager.getConnection(url, username, password);
				datasource.push(conn);
			} catch (SQLException e) {
				log.error("创建数据库连接异常:{}", e.getMessage(), e);
			}

		}
	}

	public synchronized Connection getConnection() {
		while (datasource.size() == 0) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				log.error(e.getMessage(), e);
			}
		}
		return datasource.poll();
	}

	/**
	 * 执行增删改SQL语句
	 *
	 * @param sql
	 * @param params
	 * @return
	 */
	public int executeUpdate(String sql, Object[] params) {
		int rnt = 0;
		Connection conn = null;
		PreparedStatement pstm = null;

		try {
			conn = getConnection();
			pstm = conn.prepareStatement(sql);
			for (int i = 0; i < params.length; i++) {
				pstm.setObject(i + 1, params[i]);
			}
			rnt = pstm.executeUpdate();
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			//把连接还回线程池
			if (conn != null) {
				datasource.push(conn);
			}
		}
		return rnt;
	}

	/**
	 * 执行SQL查询
	 *
	 * @param sql
	 * @param params
	 * @param queryCallBack
	 */
	public void executeQuery(String sql, Object[] params, QueryCallBack queryCallBack) {
		Connection conn = null;
		PreparedStatement pstm = null;
		ResultSet rs = null;
		try {
			conn = getConnection();
			pstm = conn.prepareStatement(sql);
			for (int i = 0; i < params.length; i++) {
				pstm.setObject(i + 1, params[i]);
			}
			rs = pstm.executeQuery();
			//查询结果回调
			queryCallBack.process(rs);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (conn != null) {
				datasource.push(conn);
			}
		}
	}

	public int[] executeBatch(String sql, List<Object[]> paramsList) {
		int[] rtn = null;
		Connection conn = null;
		PreparedStatement pstm = null;

		try {
			conn = getConnection();
			//取消自动提交
			conn.setAutoCommit(false);
			pstm = conn.prepareStatement(sql);
			for (Object[] params : paramsList) {
				for (int i = 0; i < params.length; i++) {
					pstm.setObject(i + 1, params[i]);
				}
				//加入批量的SQL参数
				pstm.addBatch();
			}
			//执行批量SQL语句
			rtn = pstm.executeBatch();
			//提交批量SQL语句
			conn.commit();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (conn != null) {
				datasource.push(conn);
			}
		}
		return rtn;
	}

	/**
	 * 查询回调函数
	 */
	public interface QueryCallBack {
		/**
		 * 处理查询结果
		 *
		 * @param rs
		 */
		void process(ResultSet rs) throws Exception;
	}
}
